package com.code.soulnetty.modbus.biz;

import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 线程池创建
 */
@Slf4j
public class ThreadPoolUtils {

    public static final Map<String, ExecutorService> POOLS = new ConcurrentHashMap<>();
    public static final String COMMON_KEY = "common";
    public static final String CLEAR_KEY = "clear";
    public static final String COLL_KEY = "coll";
    public static final String TASK_KEY = "coll";
    public static final String HEARTBEAT_KEY = "heartbeat";
    public static final String BUS_KEY = "bus";

    private ThreadPoolUtils() {
    }

    /**
     * 公共处理线程
     */
    public static ExecutorService common() {
        ExecutorService service = POOLS.get(COMMON_KEY);
        if (service != null) {
            return service;
        }
        synchronized (POOLS) {
            return POOLS.computeIfAbsent(COMMON_KEY, d -> new ThreadPoolExecutor(1, 4, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
                AtomicLong index = new AtomicLong(0);

                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "COMMON_EXECUTOR_" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
            }, new ThreadPoolExecutor.DiscardOldestPolicy()));
        }
    }

    /**
     * 清理缓存线程
     */
    public static ScheduledExecutorService clear() {
        ExecutorService service = POOLS.get(CLEAR_KEY);
        if (service != null) {
            return (ScheduledExecutorService) service;
        }
        synchronized (POOLS) {
            return (ScheduledExecutorService) POOLS.computeIfAbsent(CLEAR_KEY, d -> new ScheduledThreadPoolExecutor(1,
                    r -> {
                        Thread thread = new Thread(r, "CLEAR_EXECUTOR");
                        thread.setDaemon(true);
                        return thread;
                    }));
        }
    }

    /**
     * 采集任务
     */
    public static ExecutorService coll() {
        ExecutorService service = POOLS.get(COLL_KEY);
        if (service != null) {
            return service;
        }
        synchronized (POOLS) {
            return POOLS.computeIfAbsent(COLL_KEY, d -> new ThreadPoolExecutor(2, 2, 30, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(), new ThreadFactory() {
                AtomicLong index = new AtomicLong(0);

                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "COLL_TASK_" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
            }, (r, executor) -> log.error("COLL_EXECUTOR rejected")));
        }
    }

    /**
     * 采集周期任务线程
     */
    public static ScheduledExecutorService task() {
        ExecutorService service = POOLS.get(TASK_KEY);
        if (service != null) {
            return (ScheduledExecutorService) service;
        }
        synchronized (POOLS) {
            return (ScheduledExecutorService) POOLS.computeIfAbsent(TASK_KEY, d -> new ScheduledThreadPoolExecutor(1,
                    r -> {
                        Thread thread = new Thread(r, "TASK");
                        thread.setDaemon(true);
                        return thread;
                    }));
        }
    }

    /**
     * 采集周期任务线程
     */
    public static ScheduledExecutorService heartbeat() {
        ExecutorService service = POOLS.get(HEARTBEAT_KEY);
        if (service != null) {
            return (ScheduledExecutorService) service;
        }
        synchronized (POOLS) {
            return (ScheduledExecutorService) POOLS.computeIfAbsent(HEARTBEAT_KEY,
                    d -> new ScheduledThreadPoolExecutor(4,
                    r -> {
                        Thread thread = new Thread(r, "HEARTBEAT_VERSION");
                        thread.setDaemon(true);
                        return thread;
                    }));
        }
    }

    public static ExecutorService bus() {
        ExecutorService service = POOLS.get(BUS_KEY);
        if (service != null) {
            return service;
        }
        synchronized (POOLS) {
            return POOLS.computeIfAbsent(BUS_KEY, d -> new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
                    32, 30,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(),
                    new ThreadFactory() {
                        AtomicLong index = new AtomicLong(0);

                        @Override
                        public Thread newThread(Runnable r) {
                            return new Thread(r, "BUS_" + index.incrementAndGet());
                        }
                    }, (r, executor) -> log.error("BUS rejected")));
        }
    }
}
